package club.monkeywood.ad.dmp.report

import java.util.Properties

import club.monkeywood.ad.dmp.util.RptUtils
import com.typesafe.config.ConfigFactory
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}


/**
	* 广告在在每个地域的投放情况统计
	*
	* 实现方式---Spark Core
	* 输入：parquet文件
	* 输出：文本
	*/
object AreaAnalyseRptByRDDWithParquet {

	def main(args: Array[String]): Unit = {

		// 0 校验参数个数
		if (args.length != 2) {
			println(
				"""
					|club.monkeywood.ad.dmp.report.AreaAnalyseRptByRDDWithParquet
					|参数：
					| logInputPath
					| resultOutputPath
				""".stripMargin)
			sys.exit()
		}

		// 1 接受程序参数
		val Array(logInputPath,resultOutputPath) = args

		// 2 创建sparkconf->SparkSession
		val sparkConf = new SparkConf()
		sparkConf.setAppName(s"${this.getClass.getSimpleName}")
		sparkConf.setMaster("local[*]")
		//使用KryoSerializer更快
		// RDD 序列化到磁盘 worker与worker之间的数据传输
		sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

		val ss = SparkSession.builder()
			.appName(s"${this.getClass.getSimpleName}")
			.config(sparkConf)
			.getOrCreate()

		// 读取数据 --》parquet文件
		val df: DataFrame = ss.read.parquet(logInputPath)

		val rdd: RDD[Row] = df.rdd

		val ret = rdd.map(row => {
			// 是不是原始请求，有效请求，广告请求 List(原始请求，有效请求，广告请求)
			val reqMode = row.getAs[Int]("requestmode")
			val prcNode = row.getAs[Int]("processnode")
			// 参与竞价, 竞价成功  List(参与竞价，竞价成功, 消费, 成本)
			val effTive = row.getAs[Int]("iseffective")
			val bill = row.getAs[Int]("isbilling")
			val bid = row.getAs[Int]("isbid")
			val orderId = row.getAs[Int]("adorderid")
			val win = row.getAs[Int]("iswin")
			val winPrice = row.getAs[Double]("winprice")
			val adPayMent = row.getAs[Double]("adpayment")

			val reqList = RptUtils.caculateReq(reqMode, prcNode)
			val rtbList = RptUtils.caculateRtb(effTive, bill, bid, orderId, win, winPrice, adPayMent)
			val showClickList = RptUtils.caculateShowClick(reqMode, effTive)

			// 返回元组:((省,市),(9个指标))
			// List拼接：List++List返回一个新List
			((row.getAs[String]("provincename"), row.getAs[String]("cityname")), reqList ++ rtbList ++ showClickList)

		})

		// zip操作：
		// 输入：list1:9个元素,list2:9个元素
		// 输出：新list，包含9个元素
		// ( (list1[0],list2[0]), (list1[1],list2[1]) ... (list1[8],list2[8]) )
		// mkString(","):用“，”作为分隔符，将list转字符串
		ret.reduceByKey((list1, list2) => {
			list1.zip(list2)
				.map(t => t._1 + t._2)
		}).map(t => t._1._1+","+t._1._2+","+t._2.mkString(","))
			.saveAsTextFile(resultOutputPath)

		ss.stop()
	}

}
